<?php

namespace Core\Common\Base;

use \RdKafka\Producer;
use \RdKafka\KafkaConsumer;
use \RdKafka\Conf;
use \RdKafka\TopicConf;
/**
 * Description of KafkaHandler
 *
 * @author baiyang@ibantang.com
 */
class Kafka
{
    /**
     * 配置名
     * @var string
     */
    private $name;
    /**
     * @var \RdKafka\Producer
     */
    private $producer = null;
    /**
     * @var \RdKafka\KafkaConsumer
     */
    private $consumer = null;

    private static $instances = [];

    private function __construct($name)
    {
        $this->name = $name;
    }
    /**
     * @return \self
     */
    public static function getInstance($name)
    {
        if (empty(self::$instances[$name])) {
            self::$instances[$name] = new self($name);
        }
        return self::$instances[$name];
    }
    /**
     * 初始化生产者
     */
    private function initProducer()
    {
        $config = Config::config('kafka.'.$this->name);
        $conf = new Conf();
        if (!empty($config['sasl']) && is_array($config['sasl']) && !empty($config['sasl']['security_protocol'])) {
            foreach ($config['sasl'] as $k => $v) {
                if ('security_protocol' == $k || 'ssl_ca_location' == $k) {
                    $newKey = str_replace('_', '.', $k);
                } else {
                    $newKey = 'sasl.'.str_replace('_', '.', $k);
                }
                $conf->set($newKey, $v);
            }
        }
        $this->producer = new Producer($conf);
        $this->producer->addBrokers($config['brokers']);
    }

    /**
     *
     * @param string $topicName
     * @param mixed $payload
     * @param int $partition
     * @param string $key
     */
    public function produce($topicName, $payload, $key = null, $partition = RD_KAFKA_PARTITION_UA)
    {
        if (is_null($this->producer)) {
            $this->initProducer();
        }
        $topic = $this->producer->newTopic($topicName);
        $topic->produce($partition, 0, $payload, $key);

        // https://github.com/arnaud-lb/php-rdkafka#performance--low-latency-settings
        $times = 0;
        while ($this->producer->getOutQLen() > 0) {
            $this->producer->poll(10);
            if( $times++ >= 20){
                break;
            }
        }
    }

    /**
     * 初始化消费者
     */
    private function initConsumer()
    {
        $config = Config::config('kafka.'.$this->name);
        $kafkaConfig = [];
        $extraPath = $config['extra_config'] ?? '';
        if ($extraPath ) {
            $kafkaConfig = parse_ini_file(CORE_CONFIG_PATH . $extraPath, true);
        }
        $conf = new Conf();
        $conf->set('group.id', $config['group']);
        $conf->set('metadata.broker.list', $config['brokers']);
        if (isset($kafkaConfig['topic'])) {
            foreach ($kafkaConfig['topic'] as $k => $v) {
                $conf->set($k, $v);
            }
        }
        $topicConf = new TopicConf();
        $autoOffset = $config['offset'] ?? 'smallest';
        $topicConf->set('auto.offset.reset', $autoOffset);
        if (isset($kafkaConfig['topic_conf'])) {
            foreach ($kafkaConfig['topic_conf'] as $k => $v) {
                $topicConf->set($k, $v);
            }
        }
        if (!empty($config['sasl']) && is_array($config['sasl']) && !empty($config['sasl']['security_protocol'])) {
            foreach ($config['sasl'] as $k => $v) {
                if ('security_protocol' == $k || 'ssl_ca_location' == $k) {
                    $newKey = str_replace('_', '.', $k);
                } else {
                    $newKey = 'sasl.'.str_replace('_', '.', $k);
                }
                $conf->set($newKey, $v);
            }
        }
        $conf->setDefaultTopicConf($topicConf);
        $conf->setRebalanceCb([$this, 'rebalanceCallback']);
        $this->consumer = new KafkaConsumer($conf);
        $topics = explode(',', $config['topics']);
        $this->consumer->subscribe(array_map('trim', $topics));

    }

    public function rebalanceCallback(KafkaConsumer $kafka, $err, array $partitions = null)
    {
        switch ($err) {
            case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
                $kafka->assign($partitions);
                break;
             case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
                 $kafka->assign(null);
                 break;
             default:
                throw new \Exception($err);
        }
    }

    public function subscribe(...$topics)
    {
        if (is_null($this->consumber)) {
            $this->initConsumer();
        }
        $this->consumer->subscribe($topics);
    }
    /**
     * 获取一条消息, 开始消费
     * @param int $timeout_milli 过期时间, 毫秒
     * @return object 正确时返回
     * @return null 没有消息
     * @return boolean false 超时
     * @throws \Exception 出错
     */
    public function consume($timeout_milli = 1000)
    {
        if (is_null($this->consumer)) {
            $this->initConsumer();
        }
        $message = $this->consumer->consume($timeout_milli);
        switch ($message->err) {
            case RD_KAFKA_RESP_ERR_NO_ERROR:
                return $message;//object(RdKafka\Message)
            case RD_KAFKA_RESP_ERR__PARTITION_EOF:
                return null;//no more message
            case RD_KAFKA_RESP_ERR__TIMED_OUT:
                return false;//time out
            default:
                throw new \Exception($message->errstr(), $message->err);
        }
    }
}
